/**
  * @author marinapopova
  * Feb 24, 2016
 */
package org.elasticsearch.kafka.indexer.service.impl.examples;

import java.io.IOException;
import java.util.Set;

import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.kafka.indexer.exception.IndexerESNotRecoverableException;
import org.elasticsearch.kafka.indexer.exception.IndexerESRecoverableException;
import org.elasticsearch.kafka.indexer.service.ElasticSearchBatchService;
import org.elasticsearch.kafka.indexer.service.IMessageHandler;
import org.elasticsearch.kafka.indexer.utils.UnicodeUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import com.google.common.collect.Sets;

/**
 * 
 * This is an example of a customized Message Handler - by implementing IMessageHandler interface
 * and using the ElasticSearchBatchService to delegate most of the non-customized logic
 *
 */
public class DispatchDetailHandlerImpl implements IMessageHandler {
//	private static final Logger logger = LoggerFactory.getLogger(SimpleMessageHandlerImpl.class);
	@Autowired
	private ElasticSearchBatchService elasticSearchBatchService = null;
	@Value("${esIndexName:my_index}")
	private String indexName;
	@Value("${esIndexType:varnish}")
	private String indexType;

	@Override
	public byte[] transformMessage(byte[] inputMessage, Long offset) throws Exception {
		// do not do any transformations for this scenario - just return the message as is
		return inputMessage;
	}

	@Override
	public void addMessageToBatch(byte[] inputMessage, Long offset) throws Exception {
		String eventUUID = null; // we don't need a UUID for this simple scenario
		String routingValue = null; // we don't need routing for this simple scenario		
		elasticSearchBatchService.addEventToBulkRequest(parseToJson(inputMessage), indexName, indexType, eventUUID, 
				routingValue);
	}

	@Override
	public boolean postToElasticSearch() throws InterruptedException, IndexerESRecoverableException, IndexerESNotRecoverableException {
		return elasticSearchBatchService.postToElasticSearch();
	}
	
	public static final String[] fieldNames = new String[]{"datetime","null1","service_order_id","round","batch","flag","driver_id","distance","dispatch_time","dispatch_lat","dispatch_lng","dispatch_total_rate","dispatch_snapshot","response_time","accept_status","response_lat","response_lng","response_distance","response_time_length","decision_time","decision_total_rate","decision_result","decision_failure_reason","decision_msg_snapshot","subtract_amount","add_price_set","response_snapshot","is_assigned","route_distance","route_time_length","distance_time_length"};
	public static final Set<Integer> GEOPOINT = Sets.newHashSet(9,15);
	public static String parseToJson(byte[] inputMessage) throws IOException {
		String line = UnicodeUtils.convertUnicode(inputMessage).trim();
		String[] results = new String[fieldNames.length];
		//log分割成13份, 取出前12个字段
		String[] fields = line.split(" ",13);
		for(int i=0; i< 12; i++){
			results[i] = fields[i];
		}
		//继续将第13份分割成2份,取出 dispatch_snapshot(第13个)字段
		fields = fields[12].split("} ", 2);
		results[12] = fields[0]+"}";
		//继续分割剩下的字段
		fields = fields[1].split(" ", -1);
		for(int i=0; i< fields.length; i++){
			results[i+13] = fields[i];
		}
		return buildJson(results);
	}

	public static String buildJson(String[] fields) throws IOException{
		XContentBuilder jsonBuild = XContentFactory.jsonBuilder();
		jsonBuild.startObject();
		for(int i=0; i< fieldNames.length; i++){
			if(i==0){
				jsonBuild.field(fieldNames[i],fields[i]+"+0800");
			}else if(GEOPOINT.contains(i)){
				jsonBuild.field(fieldNames[i], fields[i] + "," + fields[++i]);
			}
			else{
				jsonBuild.field(fieldNames[i],fields[i]);
			}
		}
		jsonBuild.endObject();
		return jsonBuild.string();
	}
	
	public static String parseToJson_bak(byte[] inputMessage) throws IOException{
		String line = UnicodeUtils.convertUnicode(inputMessage);
		//使用XContentBuilder创建json数据
		XContentBuilder jsonBuild = XContentFactory.jsonBuilder();
		jsonBuild.startObject();
		
		//log分割成13份,取出前12个字段
		String[] fields = line.split(" ",13);
		for(int i=0; i< 12; i++){
			if(i==0){
				jsonBuild.field(fieldNames[i],fields[i]+"+0800");
			}else{
				jsonBuild.field(fieldNames[i],fields[i]);
			}
		}
		//继续将第13份分割成2份,取出 dispatch_snapshot(第13个)字段
		fields = fields[12].split("} ", 2);
		jsonBuild.field(fieldNames[12],fields[0]+"}");
		//继续分割剩下的字段
		fields = fields[1].split(" ", -1);
		for(int i=0; i< fields.length; i++){
			jsonBuild.field(fieldNames[i+13],fields[i]);
		}
		jsonBuild.endObject();
		return jsonBuild.string();
	}
	
	public static void main(String[] args) throws IOException {
		String log="2016-11-02_00:03:25  6348030879597585909 1 1 0 491278 1435 1478016115 31.195845269097 121.69832790799 9800 {\"is_self_employed\":0,\"country\":\"CN\",\"driver_id\":\"491278\",\"flag\":61454,\"base_score\":20,\"driver_type\":5,\"color\":4,\"distance\":1435,\"city\":\"bj\",\"latitude\":31.195845269097,\"total_rate\":9800,\"device_type\":\"2\",\"audit_status\":20,\"has_logistics\":0,\"discount_rate\":100,\"distance_time_length\":430.5,\"contribution_rate\":26,\"last_position_time\":1478016111,\"surpport_face_pay\":1,\"evaluation\":960,\"evaluation_rate\":96,\"has_qualified\":0,\"contribution\":20000,\"cellphone\":\"18916167612\",\"silence_end_at\":1467655419,\"work_status\":0,\"car_id\":519863,\"brand\":\"荣威 350\",\"seat_num\":0,\"route_distance\":0,\"longitude\":121.69832790799,\"car_type_id\":37,\"route_time_length\":0,\"device_id\":632717,\"base_score_rate\":20,\"good_comment_rate\":96,\"driver_level\":1,\"version\":\"204\",\"car_brand_id\":106,\"car_model_id\":1046,\"vehicle_number\":\"皖SJD566\",\"name\":\"戴军弟\",\"imei\":\"5fce7e3d0b50be699136daf56dece04e\",\"distance_rate\":98,\"is_assigned\":0,\"add_price_set\":{\"add_price_redispatch\":0,\"add_price_rate\":0,\"add_price_max_amount\":10000000,\"add_price_type\":0,\"add_price_vip\":0,\"strategy_id\":0,\"add_total_amount\":0,\"total_add_price_rate\":0,\"bidding_id\":\"6348030876911017356\",\"total_magnification\":0,\"bidding_price\":0,\"add_amount_str\":\"\",\"add_amount_str_full\":\"\"}} 0 0 0 0 0 0 0 0 0 0 - 0 {\"add_price_redispatch\":0,\"add_price_rate\":0,\"add_price_max_amount\":10000000,\"add_price_type\":0,\"add_price_vip\":0,\"strategy_id\":0,\"add_total_amount\":0,\"total_add_price_rate\":0,\"bidding_id\":\"6348030876911017356\",\"total_magnification\":0,\"bidding_price\":0,\"add_amount_str\":\"\",\"add_amount_str_full\":\"\"} - 0 0 0 430.5";
		System.out.println(parseToJson(log.getBytes()));
	}
}
